import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.DStream

object WordCountToMySQL {
  def main(args: Array[String]): Unit = {
    // Spark配置
    val conf = new SparkConf()
      .setAppName("WordCountToMySQL")
      .setMaster("local[2]")

    // 创建StreamingContext，批次间隔5秒
    val ssc = new StreamingContext(conf, Seconds(5))

    // 从socket源创建DStream，监听本地9999端口
    val lines = ssc.socketTextStream("localhost", 9999)

    // 处理数据：拆分单词并计数
    val wordCounts = lines.flatMap(_.split(" "))
      .filter(_.nonEmpty)
      .map(word => (word, 1))
      .reduceByKey(_ + _)

    // 每个批次统计出现次数最高的前5名单词
    wordCounts.foreachRDD { rdd =>
      val sortedWords = rdd.sortBy(_._2, ascending = false).take(5)
      saveToMySQL(sortedWords)
    }

    // 启动流计算
    ssc.start()
    ssc.awaitTermination()
  }

  // 将统计结果保存到MySQL数据库
  def saveToMySQL(sortedWords: Array[(String, Int)]): Unit = {
    var connection: Connection = null
    var preparedStatement: PreparedStatement = null

    try {
      // MySQL连接参数
      val url = "jdbc:mysql://localhost:3306/spark_db"
      val username = "root"
      val password = "123456"

      // 加载JDBC驱动
      Class.forName("com.mysql.jdbc.Driver")

      // 建立连接
      connection = DriverManager.getConnection(url, username, password)

      // 准备SQL语句（使用ON DUPLICATE KEY UPDATE实现更新）
      val sql = """
        INSERT INTO word_counts (word, count)
        VALUES (?, ?)
        ON DUPLICATE KEY UPDATE count = count + VALUES(count)
      """
      preparedStatement = connection.prepareStatement(sql)

      // 遍历前5名单词并插入/更新数据库
      sortedWords.foreach { case (word, count) =>
        preparedStatement.setString(1, word)
        preparedStatement.setInt(2, count)
        preparedStatement.executeUpdate()
      }
    } catch {
      case e: Exception => e.printStackTrace()
    } finally {
      // 关闭资源
      if (preparedStatement != null) preparedStatement.close()
      if (connection != null) connection.close()
    }
  }
}